package com.xian.java.stream;

import com.xian.bean.Temperature;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

public class FlatMapTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //DataStreamSource<String> stringDataStreamSource = env.readTextFile("E:/test/Temperature.txt");
        DataStream<String> dataStream = env.readTextFile("E:/test/Temperature.txt"); //用父类接
        //dataStream.print();


        DataStream<Temperature> temperatureDataStream = dataStream.flatMap(new RichFlatMapFunction<String, Temperature>() {
            @Override
            public void flatMap(String line, Collector<Temperature> out) throws Exception {
                String[] arr = line.split(",");
                int id = Integer.parseInt(arr[0]);
                double temperature = Double.parseDouble(arr[1]);
                long timestamp = Long.parseLong(arr[2]); // 1478247229842  13位毫秒值
                Temperature t = new Temperature(id, temperature, timestamp);

                out.collect(t);
            }
        });

        temperatureDataStream.print();




        env.execute();


    }
}
